package com.ruyuan.eshop.data.migrate.tool.controller;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.ruyuan.eshop.data.migrate.tool.canal.binlog.BinlogConsumer;
import com.ruyuan.eshop.data.migrate.tool.config.prop.CanalServerProperties;
import com.ruyuan.eshop.data.migrate.tool.core.JsonResult;
import com.ruyuan.eshop.data.migrate.tool.thread.BinlogConsumerThread;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.net.InetSocketAddress;

/**
 * 增量数据同步处理器
 * @author zhonghuashishan
 * @version 1.0
 */
@RestController
public class IncrementDataSyncController {

    @Autowired
    private CanalServerProperties canalServerProperties;

    @Autowired
    private BinlogConsumer binlogConsumer;

    /**
     * 开启增量数据同步
     *
     * 说明：一定要等到历史数据同步完之后，才能触发开启增量数据的同步
     *
     * @return
     */
    @GetMapping("/startIncrementDataSync")
    public JsonResult<Boolean> startIncrementDataSync() {

        // 防止重复执行，在本地数据库记录一个开关 TODO

        String hostname = canalServerProperties.getHostname();
        int port = canalServerProperties.getPort();
        String destination = canalServerProperties.getDestination();
        String username = canalServerProperties.getUsername();
        String password = canalServerProperties.getPassword();

        CanalConnector canalConnector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(hostname, port), destination, username, password);
        // 建立连接
        canalConnector.connect();

        // 异步线程开启canal客户端
        Thread thread = new Thread(new BinlogConsumerThread(canalConnector, binlogConsumer));
        thread.setDaemon(true);
        thread.start();

        return JsonResult.buildSuccess(true);
    }
}